package com.luolc;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.example.util.Runner;
import io.vertx.ext.web.FileUpload;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
import io.vertx.ext.web.handler.BodyHandler;
import okhttp3.Interceptor;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okhttp3.internal.http.HttpMethod;

import java.io.IOException;

/**
 * @author LuoLiangchen
 */
public class ProxyService extends AbstractVerticle {
    private static final String ADDRESS = "10.128.202.205";

    private static final MediaType JSON = MediaType.parse("application/json; charset=utf-8");

    private static final MediaType TIFF = MediaType.parse("image/tiff");

    private static final String HEADER_LOCATION = "Location";

    public static void main(String[] args) {
        Runner.runExample(ProxyService.class);
    }

    @Override
    public void start() throws Exception {
        final Router router = Router.router(vertx);

        router.route().handler(BodyHandler.create().setUploadsDirectory("uploads"));
        router.route().handler(this::transfer);

        vertx.createHttpServer().requestHandler(router::accept).listen(10000);
    }

    private void transfer(RoutingContext context) {
        final String uri = context.request().absoluteURI();
        final String targetUrl =
                "http://" + ADDRESS + ":50700/" + uri.substring(uri.indexOf("webhdfs/v1"));
        final HttpServerResponse response = context.response();
        final String method = context.request().method().toString();
        final OkHttpClient client = new OkHttpClient.Builder()
                .addNetworkInterceptor(new RedirectInterceptor())
                .build();

        final RequestBody requestBody;
        if (!HttpMethod.permitsRequestBody(method)) {
            requestBody = null;
        } else {
            requestBody = RequestBody.create(JSON, "");
        }

        final Request transferRequest = new Request.Builder()
                .url(targetUrl)
                .method(method, requestBody)
                .build();
        try {
            final Response transferResponse =
                    uploadFileOrNot(context, client, client.newCall(transferRequest).execute());
            response.headers().clear();
            for (String name : transferResponse.headers().names()) {
                if (!"Transfer-Encoding".equals(name)) {
                    final String value = transferResponse.header(name);
                    response.putHeader(name, value);
                }
            }
            // CORS
            response.putHeader("Access-Control-Allow-Origin", "*");
            response.putHeader("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, HEAD, OPTIONS");

            final ResponseBody body = transferResponse.body();
            if (body == null) {
                response.end();
            } else {
                response.end(Buffer.buffer(body.bytes()));
            }
        } catch (IOException ex) {
            response.setStatusCode(500);
        }
    }

    private Response uploadFileOrNot(RoutingContext context, OkHttpClient client, Response origin)
            throws IOException {
        final String location = origin.header(HEADER_LOCATION);
        if (location != null && location.contains("op=CREATE")) {
            final FileUpload fileUpload = context.fileUploads().iterator().next();
            final byte[] fileBytes = vertx.fileSystem().readFileBlocking(fileUpload.uploadedFileName()).getBytes();
            final RequestBody requestBody = RequestBody.create(TIFF, fileBytes);
            final Request request = new Request.Builder()
                    .url(location)
                    .method(context.request().method().toString(), requestBody)
                    .build();
            return client.newCall(request).execute();
        } else {
            return origin;
        }
    }

    private static class RedirectInterceptor implements Interceptor {
        @Override
        public Response intercept(Chain chain) throws IOException {
            final Request request = chain.request();
            final Response response = chain.proceed(request);
            final String location = response.header(HEADER_LOCATION);
            final Response newResponse;
            if (location != null) {
                newResponse = response.newBuilder()
                        .header(HEADER_LOCATION, location.replaceAll("yourandell", ADDRESS))
                        .build();
            } else {
                newResponse = response;
            }
            return newResponse;
        }
    }
}
